use anyhow::Result;
use chrono::DateTime;
use chrono::Local;
use eventsource_client::Client;
use eventsource_client::Event;
use futures::TryStreamExt;
use once_cell::sync::Lazy;
use serde::Deserialize;
use serde::Serialize;
use serde_json::Value;
use std::sync::Arc;
use std::collections::HashMap;
use std::time::Duration;
use tokio::sync::mpsc::Sender;

use super::EastmoneyTick;

// 定义静态全局变量stockMessage
pub static EASTMONEY_FIELDS: Lazy<HashMap<&str, &str>> = Lazy::new(|| {
    let mut map: HashMap<&str, &str> = HashMap::new();
    map.insert("f2", "最新价");
    map.insert("f3", "涨跌幅");
    map.insert("f4", "涨跌额");
    map.insert("f5", "总手");
    map.insert("f6", "成交额");
    map.insert("f7", "振幅");
    map.insert("f8", "换手率");
    map.insert("f9", "市盈率");
    map.insert("f10", "量比");
    map.insert("f11", "5分钟涨跌幅");
    map.insert("f12", "股票代码");
    map.insert("f13", "市场");
    map.insert("f14", "股票名称");
    map.insert("f15", "最高价");
    map.insert("f16", "最低价");
    map.insert("f17", "开盘价");
    map.insert("f18", "昨收");
    map.insert("f20", "总市值");
    map.insert("f21", "流通市值");
    map.insert("f22", "涨速");
    map.insert("f23", "市净率");
    map.insert("f24", "60日涨跌幅");
    map.insert("f25", "年初至今涨跌幅");
    map.insert("f26", "上市日期");
    map.insert("f28", "昨日结算价");
    map.insert("f30", "现手");
    map.insert("f31", "买入价");
    map.insert("f32", "卖出价");
    map.insert("f33", "委比");
    map.insert("f34", "外盘");
    map.insert("f35", "内盘");
    map.insert("f36", "人均持股数");
    map.insert("f37", "净资产收益率(加权)");
    map.insert("f38", "总股本");
    map.insert("f39", "流通股");
    map.insert("f40", "营业收入");
    map.insert("f41", "营业收入同比");
    map.insert("f42", "营业利润");
    map.insert("f43", "投资收益");
    map.insert("f44", "利润总额");
    map.insert("f45", "净利润");
    map.insert("f46", "净利润同比");
    map.insert("f47", "未分配利润");
    map.insert("f48", "每股未分配利润");
    map.insert("f49", "毛利率");
    map.insert("f50", "总资产");
    map.insert("f51", "流动资产");
    map.insert("f52", "固定资产");
    map.insert("f53", "无形资产");
    map.insert("f54", "总负债");
    map.insert("f55", "流动负债");
    map.insert("f56", "长期负债");
    map.insert("f57", "资产负债比率");
    map.insert("f58", "股东权益");
    map.insert("f59", "股东权益比");
    map.insert("f60", "公积金");
    map.insert("f61", "每股公积金");
    map.insert("f62", "主力净流入");
    map.insert("f63", "集合竞价");
    map.insert("f64", "超大单流入");
    map.insert("f65", "超大单流出");
    map.insert("f66", "超大单净额");
    map.insert("f69", "超大单净占比");
    map.insert("f70", "大单流入");
    map.insert("f71", "大单流出");
    map.insert("f72", "大单净额");
    map.insert("f75", "大单净占比");
    map.insert("f76", "中单流入");
    map.insert("f77", "中单流出");
    map.insert("f78", "中单净额");
    map.insert("f81", "中单净占比");
    map.insert("f82", "小单流入");
    map.insert("f83", "小单流出");
    map.insert("f84", "小单净额");
    map.insert("f87", "小单净占比");
    map.insert("f88", "当日DDX");
    map.insert("f89", "当日DDY");
    map.insert("f90", "当日DDZ");
    map.insert("f91", "5日DDX");
    map.insert("f92", "5日DDY");
    map.insert("f94", "10日DDX");
    map.insert("f95", "10日DDY");
    map.insert("f97", "DDX飘红天数(连续)");
    map.insert("f98", "DDX飘红天数(5日)");
    map.insert("f99", "DDX飘红天数(10日)");
    map.insert("f100", "行业");
    map.insert("f101", "板块领涨股");
    map.insert("f102", "地区板块");
    map.insert("f103", "备注");
    map.insert("f104", "上涨家数");
    map.insert("f105", "下跌家数");
    map.insert("f106", "平家家数");
    map.insert("f112", "每股收益");
    map.insert("f113", "每股净资产");
    map.insert("f114", "市盈率（静）");
    map.insert("f115", "市盈率（TTM）");
    map.insert("f124", "当前交易时间");
    map.insert("f128", "板块领涨股");
    map.insert("f129", "净利润");
    map.insert("f130", "市销率TTM");
    map.insert("f131", "市现率TTM");
    map.insert("f132", "总营业收入TTM");
    map.insert("f133", "股息率");
    map.insert("f134", "行业板块的成分股数");
    map.insert("f135", "净资产");
    map.insert("f138", "净利润TTM");
    map.insert("f221", "更新日期");
    map.insert("f400", "pre：盘前时间 after：盘后时间 period：盘中时间");

    map // 返回填充后的HashMap
});

/// Mapping to [STOCK_600519_pankou.json](file:///Users/CC/gitee/conanchen/fuquants-workspace/crates/fudacli/eastmoney/sample/STOCK_600519_pankou.json)
#[derive(Debug, Serialize, Deserialize, Clone, Default)]
pub struct StockTransaction {
    pub rc: i32,
    pub rt: i32,
    pub svr: u64,
    pub lt: i32,
    pub full: i32,
    pub dlmkts: String,
    pub data: StockTransactionData,
}

#[derive(Debug, Serialize, Deserialize, Clone, Default)]
pub struct StockTransactionData {
    pub code: Option<Value>,
    pub market: Option<Value>,
    pub decimal: Option<Value>,
    #[serde(rename = "prePrice")]
    pub pre_price: Option<Value>,
    /// 时间，价格，成交量，成交笔数，卖1买2竞4
    /// "09:15:05,1448.00,3,0,4",
    /// "09:34:05,1445.62,12,10,2",
    /// "09:34:08,1445.61,5,5,1",
    /// "09:34:11,1446.00,15,11,2",
    /// "09:34:14,1445.92,1,1,1",
    pub details: Option<Vec<String>>,
}
impl StockTransactionData {
    pub fn ticks(self) -> Option<Vec<EastmoneyTick>> {
        if let Some(ts) = &self.details {
            let ticks: Vec<EastmoneyTick> = ts
                .iter()
                .map(|data| {
                    let parts: Vec<&str> = data.split(',').collect();
                    let time_: DateTime<Local> = dateparser
                        ::parse(parts[0])
                        .unwrap_or_default()
                        .into();
                    EastmoneyTick {
                        time: time_,
                        price: parts[1].parse().unwrap_or_default(),
                        volume: parts[2].parse().unwrap_or_default(),
                        trade_count: parts[3].parse().unwrap_or_default(),
                        s1b2j4: parts[4].parse().unwrap_or_default(),
                    }
                })
                .collect();
            Some(ticks)
        } else {
            None
        }
    }
}

#[derive(Debug)]
pub struct StockTransactionEvent {
    pub time: u64,
    pub code: String,
    pub event: Option<Event>,
    pub success: bool,
}
impl StockTransaction {
    /// 分时成交
    /// Refer : https://quote.eastmoney.com/sh513130.html
    ///
    /// Data: https://73.push2delay.eastmoney.com/api/qt/stock/details/sse?fields1=f1,f2,f3,f4&fields2=f51,f52,f53,f54,f55&mpi=1000&ut=bd1d9ddb04089700cf9c27f6f7426281&fltt=2&pos=-11&secid=1.513130&wbp2u=|0|0|0|web
    ///
    /// secid: str e.g. e.g. sh 1.600519, sz 0.000001/
    pub async fn intrady_transactions_sse(
        secid: &str,
        tx: Arc<Sender<StockTransactionEvent>>
    ) -> Result<()> {
        // GET /api/qt/stock/details/sse?fields1=f1,f2,f3,f4&fields2=f51,f52,f53,f54,f55&mpi=1000&ut=bd1d9ddb04089700cf9c27f6f7426281&fltt=2&pos=-11&secid=1.513130&wbp2u=|0|0|0|web HTTP/1.1
        // Accept: text/event-stream
        // Accept-Encoding: gzip, deflate, br, zstd
        // Accept-Language: zh-CN,zh;q=0.9,en;q=0.8
        // Cache-Control: no-cache
        // Connection: keep-alive
        // Host: 73.push2delay.eastmoney.com
        // Origin: null
        // Pragma: no-cache
        // Referer: https://quote.eastmoney.com/sh513130.html
        // Sec-Fetch-Dest: empty
        // Sec-Fetch-Mode: cors
        // Sec-Fetch-Site: same-site
        // User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36
        // sec-ch-ua: "Chromium";v="122", "Not(A:Brand";v="24", "Google Chrome";v="122"
        // sec-ch-ua-mobile: ?0
        // sec-ch-ua-platform: "macOS"

        // Query String Parameters:
        //
        // fields1: f1,f2,f3,f4
        // fields2: f51,f52,f53,f54,f55
        // mpi: 1000
        // ut: bd1d9ddb04089700cf9c27f6f7426281
        // fltt: 2
        // pos: -11
        // secid: 1.513130
        // wbp2u: |0|0|0|web

        // {"rc":0,"rt":12,"svr":182992900,"lt":2,"full":1,"dlmkts":"8,10,128","data":{"code":"513130","market":1,"decimal":3,"prePrice":0.481,"details":["10:22:55,0.486,48,2,2","10:22:58,0.485,57,1,1","10:23:01,0.486,156,3,2","10:23:04,0.486,1000,1,2","10:23:07,0.486,120,2,2","10:23:10,0.485,4059,2,1","10:23:13,0.486,100,1,2","10:23:16,0.485,3240,4,1","10:23:19,0.485,3383,4,1","10:23:22,0.486,2326,2,2","10:23:25,0.486,31,2,2"]}}
        // {"rc":0,"rt":12,"svr":182992900,"lt":2,"full":0,"dlmkts":"8,10,128","data":{"details":["10:23:28,0.486,5153,3,2","10:23:31,0.485,1040,5,1"]}}
        let host = "75.push2delay.eastmoney.com";
        let fields1 = "f1,f2,f3,f4";
        let fields2 = "f51,f52,f53,f54,f55";
        let mpi = 1000;
        let ut = utils::generate_random_hex_string(32);
        let fltt = 2;
        let pos = -11;
        // let seci= "1.513130";
        let wbp2u = "|0|0|0|web";

        let url = format!(
            "https://{host}/api/qt/stock/details/sse?fields1={fields1}&fields2={fields2}&mpi={mpi}&ut={ut}&fltt={fltt}&pos={pos}&secid={secid}&wbp2u={wbp2u}"
        );
        println!("\nData: {}", url);

        let eventsource_client = eventsource_client::ClientBuilder
            ::for_url(&url)?
            .header(reqwest::header::ACCEPT.as_str(), "text/event-stream")?
            .header("Accept-Encoding", "gzip, deflate, br")?
            .header("Accept-Language", "zh-CN,zh;q=0.9,en;q=0.8")?
            .header("Cache-Control", "no-cache")?
            .header("Connection", "keep-alive")?
            .header("Host", &format!("{host}"))
            ? // must be same as url's host
            .header("Referer", "http://quote.eastmoney.com/sh513130.html")?
            .header("Sec-Fetch-Dest", "empty")?
            .header("Sec-Fetch-Mode", "cors")?
            .header("Sec-Fetch-Site", "cross-site")?
            .header(
                "User-Agent",
                "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
            )?
            .header(
                "sec-ch-ua",
                "\"Not_A Brand\";v=\"8\", \"Chromium\";v=\"120\", \"Google Chrome\";v=\"120\""
            )?
            .header("sec-ch-ua-mobile", "?0")?
            .reconnect(
                eventsource_client::ReconnectOptions
                    ::reconnect(true)
                    .retry_initial(false)
                    .delay(Duration::from_secs(1))
                    .backoff_factor(2)
                    .delay_max(Duration::from_secs(60))
                    .build()
            )
            .build();

        let mut eventsource_stream = eventsource_client
            .stream()
            .map_ok(|event| {
                match event {
                    eventsource_client::SSE::Event(ev) => {
                        // println!("\n[{secid}] got an event: {} {}", ev.event_type, ev.data);
                        Some(ev)
                    }
                    eventsource_client::SSE::Comment(_comment) => {
                        // println!("\n[{secid}] got a comment: \n{}", comment);
                        None
                    }
                }
            })
            .map_err(|err| eprintln!("\n[{secid}] error streaming events: {:?}", err));

        while let Ok(Some(Some(evt))) = eventsource_stream.try_next().await {
            let evt_: StockTransactionEvent = StockTransactionEvent {
                time: utils::duration_ms_since_epoch(),
                code: secid.to_owned(),
                event: Some(evt.clone()),
                success: true,
            };

            println!(
                "\n>>>Sent: {:?} {} {}-{}",
                DateTime::from_timestamp_millis(evt_.time.try_into().unwrap()),
                evt_.code,
                evt.event_type,
                evt.data
            );

            let _ = tx.send(evt_).await;
        }

        println!("\n\n****pankou_sse for {secid} End caused by error!!!???\n");
        let evt_: StockTransactionEvent = StockTransactionEvent {
            time: utils::duration_ms_since_epoch(),
            code: secid.to_owned(),
            event: None,
            success: false,
        };
        let _ = tx.send(evt_).await;

        Ok(())
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use chrono::Local;
    use regex::Regex;
    use serde_json;
    use tokio::sync::mpsc::channel;

    #[tokio::test]
    async fn test_intrady_transactions_sse() {
        let symbols = vec![
            "1.600519" //"1.513130", "1.511670", "1.515880", "0.159509", "1.515050",
        ]
            .iter()
            .map(|&symbol| symbol.to_string())
            .collect::<Vec<String>>();

        let (tx, mut rx) = channel::<StockTransactionEvent>(100); // Create a channel
        let tx_shared = Arc::new(tx);

        let mut handles = vec![];

        for symbol in &symbols {
            let tx_clone = Arc::clone(&tx_shared);
            let symbol_clone = symbol.clone();
            let handle = tokio::spawn(async move {
                process_symbol(symbol_clone, tx_clone).await;
            });
            handles.push(handle);
        }

        // Process events sequentially from the shared receiver
        while let Some(event) = rx.recv().await {
            let dd: StockTransaction = serde_json
                ::from_str(&event.event.unwrap().data)
                .unwrap_or_default();
            println!("<<<Received: {} {} {:?}", event.time, event.code, dd.data.ticks());

            if event.success == false {
                break;
            }
        }

        for handle in handles {
            handle.await.unwrap(); // Handle errors if needed
        }

        println!("End of all handles");
    }

    async fn process_symbol(symbol: String, tx: Arc<Sender<StockTransactionEvent>>) {
        // Use the shared sender to send events
        let _ = StockTransaction::intrady_transactions_sse(&symbol, tx).await;
    }

    fn _remove_before_dot(s: &str) -> String {
        let re = Regex::new(r"^(.*?)\.").unwrap();
        re.replace_all(s, "").to_string()
    }

    #[tokio::test]
    async fn test_tick_struct() {
        let tick_data = vec![
            "13:46:17,0.450,269,1,2",
            "13:46:20,0.450,100,1,2"
            // "13:46:23,0.450,1216,2,2",
            // "13:46:26,0.450,8621,2,2",
            // "13:46:29,0.450,138,2,2",
            // "13:46:38,0.450,8621,2,2",
            // "13:46:41,0.449,1030,2,1",
            // "13:46:44,0.449,2011,2,1",
            // "13:46:47,0.449,565,1,1",
            // "13:46:50,0.450,30,1,2",
            // "13:46:56,0.449,330,1,1"
        ];

        let ticks: Vec<EastmoneyTick> = tick_data
            .iter()
            .map(|&data| {
                let parts: Vec<&str> = data.split(',').collect();
                let time_: DateTime<Local> = dateparser::parse(parts[0]).unwrap_or_default().into();
                println!("time: {:?}", time_);
                EastmoneyTick {
                    time: time_,
                    price: parts[1].parse().unwrap(),
                    volume: parts[2].parse().unwrap(),
                    trade_count: parts[3].parse().unwrap(),
                    s1b2j4: parts[4].parse().unwrap(),
                }
            })
            .collect();

        // Print out the tick data
        for tick in ticks {
            println!(
                "Time: {}, Price: {}, Volume: {}, Trade Num: {}, Buy/Sell Indicator: {}",
                tick.time,
                tick.price,
                tick.volume,
                tick.trade_count,
                tick.s1b2j4
            );
        }
    }
}
